package com.bw.util;

import com.alibaba.fastjson.JSONObject;
import com.bw.common.GmallConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.hbase.sink.HBaseMutationConverter;
import org.apache.flink.connector.hbase.sink.HBaseSinkFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;


public class SinkHBaseUtil extends RichSinkFunction<JSONObject> {

    //标记为 transient，表示这些字段不需要被序列化
    //maxSize 批量写入的数据量
    //写入时间间隔
    private transient Integer maxSize = 1000;
    private transient Long delayTime = 5000L;

    public SinkHBaseUtil(Integer maxSize, Long delayTime) {
        this.maxSize = maxSize;
        this.delayTime = delayTime;
    }

    //用于记录数据最后写入的时间
    private transient Long lastInvokeTime;
    //    用于存储待写入 HBase 的 Put 对象。列表的初始容量设置为 maxSize，以优化内存使用。
    private transient final List<Put> puts = new ArrayList<>(maxSize);
    public SinkHBaseUtil() {
    }

    private  static Connection connection = null;
    @Override
    public void open(Configuration parameters) throws Exception {

         connection = HBaseUtil.getConnection(
                ConfigUtil.getZookeeperQuorum()
                , ConfigUtil.getZookeeperClientPort());
        System.out.println("初始化hbase链接>>>>>>>>>");

        // 获取系统当前时间
        lastInvokeTime = System.currentTimeMillis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHbaseCon(connection);
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //获取id 当做 rowkey

        JSONObject data = value.getJSONObject("data");

        String id = data.getString("id");

        Put put = new Put(id.getBytes());

        for (Map.Entry<String, Object> entry : data.entrySet()) {
                put.addColumn("data".getBytes(), entry.getKey().getBytes(), entry.getValue().toString().getBytes());
        }

//        puts.add(put);

//        //使用ProcessingTime
//        long currentTime = System.currentTimeMillis();
//
//        System.out.println("耗时 "+(currentTime - lastInvokeTime)/1000+" 秒");
        //定义批量提交条件
//        if (puts.size() == maxSize || currentTime - lastInvokeTime >= delayTime){
            String sinkTable = value.getString("sinkTable");
            String hbaseNamespase = GmallConfig.HBASE_SCHEMA;

            TableName tableName = TableName.valueOf(hbaseNamespase + ":" + sinkTable);

            Table table = connection.getTable(tableName);
            //提交
            table.put(put);
            //清空缓存
//            puts.clear();
            //重置时间

//            lastInvokeTime = currentTime;
            table.close();
//        }

    }
}
